package client

import (
	"context"
	"fmt"
	"net"
	"sync"
	"time"

	kcplib "github.com/xtaci/kcp-go/v5"
	"golang.org/x/net/websocket"
	"google.golang.org/protobuf/proto"

	"gitee.com/hasika/common-server/codec"
	"gitee.com/hasika/common-server/conn/inet"
	"gitee.com/hasika/common-server/conn/kcp"
	"gitee.com/hasika/common-server/conn/tcp"
	"gitee.com/hasika/common-server/conn/udp"
	"gitee.com/hasika/common-server/conn/ws"
	"gitee.com/hasika/common-server/option"
	"gitee.com/hasika/gotool/logx"

	"gitee.com/hasika/common-client/callback"
)

type Req interface {
	codec.Packet
	callback.ISeqCallback
}

type Rsp interface {
	codec.RspPacket
	callback.ISeqCallback
}

type Client[T Req, H Rsp] struct {
	ctx                context.Context
	cancel             context.CancelFunc
	receivedCh         chan H
	writePacketBytesCh chan []byte
	cmr                *callback.Manager[H]
	conn               inet.Conn
	clientOption       *Option[T, H]
	remoteOption       *option.RemoteOption
	rspToProto         PostProcessor[H]
	wg                 *sync.WaitGroup
}

func (c *Client[T, H]) ClientOption() *Option[T, H] {
	return c.clientOption
}

func (c *Client[T, H]) RemoteOption() *option.RemoteOption {
	return c.remoteOption
}

func (c *Client[T, H]) Conn() inet.Conn {
	return c.conn
}

func NewClient[T Req, H Rsp](parentCtx context.Context, option *Option[T, H], remoteOption *option.RemoteOption) (*Client[T, H], error) {
	ctx, cancel := context.WithCancel(parentCtx)
	ret := &Client[T, H]{
		receivedCh:         make(chan H, option.RcvQueueSize()),
		writePacketBytesCh: make(chan []byte, option.SendQueueSize()),
		cmr:                callback.NewCallbackManager[H](ctx),
		clientOption:       option,
		remoteOption:       remoteOption,
		rspToProto:         option.GetRspPostProcessor(),
		wg:                 &sync.WaitGroup{},
	}
	c, err := NewConnection(ctx, option, remoteOption)
	if err != nil {
		cancel()
		return nil, err
	}
	ret.conn = c
	ret.ctx = ctx
	ret.cancel = cancel
	go ret.run()
	option.GetBindFunc()(ret)
	return ret, nil
}

func NewConnection[T Req, H Rsp](ctx context.Context, option *Option[T, H], remoteOption *option.RemoteOption) (inet.Conn, error) {
	network := remoteOption.Network()
	serverAddress := fmt.Sprintf("%s:%d", remoteOption.Host(), remoteOption.Port())
	writeCtx := inet.NewWriteToChanCtx()
	switch network {
	case inet.Tcp:
		conn, err := net.Dial("tcp", serverAddress)
		if err != nil {
			return nil, err
		}
		tcpN := tcp.NewTcpConnection(conn.(*net.TCPConn))
		tcpN.InitConnection(tcpN, option.GetConnectionCallback(), false, writeCtx, remoteOption.RecordReadWrite())
		return tcpN, nil
	case inet.Kcp:
		conn, err := kcplib.Dial(serverAddress)
		if err != nil {
			return nil, err
		}
		tcpN := kcp.NewKcpConnection(conn.(*kcplib.UDPSession), option.KcpOption())
		tcpN.InitConnection(tcpN, option.GetConnectionCallback(), false, writeCtx, remoteOption.RecordReadWrite())
		return tcpN, nil
	case inet.Udp:
		serverUdp, err := net.ResolveUDPAddr("udp", serverAddress)
		if err != nil {
			return nil, err
		}
		conn, err := net.DialUDP("udp", nil, serverUdp)
		if err != nil {
			return nil, err
		}
		tcpN := udp.NewClientConnection(conn)
		tcpN.InitConnection(tcpN, option.GetConnectionCallback(), false, writeCtx, remoteOption.RecordReadWrite())
		return tcpN, nil
	case inet.Websocket:
		conn, err := websocket.Dial(fmt.Sprintf("ws://%s", serverAddress), "", fmt.Sprintf("http://%s", serverAddress))
		if err != nil {
			return nil, err
		}
		tcpN := ws.NewWebsocketConnection(conn, ctx)
		tcpN.InitConnection(tcpN, option.GetConnectionCallback(), false, writeCtx, remoteOption.RecordReadWrite())
		return tcpN, nil
	}
	panic("not supported")
}

func (c *Client[T, H]) CallBodyWithCallback(message proto.Message, callback func(message proto.Message, e error)) error {
	req := c.clientOption.GetBuildReqFunc()(message)
	return c.CallRawWithCallback(req, callback)
}

func (c *Client[T, H]) CallBodyAndWait(message proto.Message) (proto.Message, error) {
	req := c.clientOption.GetBuildReqFunc()(message)
	return c.CallRawAndWait(req)
}

func (c *Client[T, H]) CallBodyOnly(message proto.Message) error {
	req := c.clientOption.GetBuildReqFunc()(message)
	return c.CallRawOnly(req)
}

func (c *Client[T, H]) CallRawWithCallback(req T, callback func(message proto.Message, e error)) error {
	c.cmr.AddCallback(req.GetCallbackSeq(), func(rsp H, e error) {
		ret := rsp.GetMessage()
		if c.rspToProto != nil {
			ret, e = c.rspToProto(rsp, e)
		}
		callback(ret, e)
	})
	return c.CallRawOnly(req)
}

func (c *Client[T, H]) CallRawAndWait(req T) (proto.Message, error) {
	promise := c.cmr.AddPromise(req.GetCallbackSeq())
	err := c.CallRawOnly(req)
	if err != nil {
		promise.Cancel()
		return nil, err
	}
	rsp, err := promise.Wait()
	if err != nil {
		return nil, err
	}
	ret := rsp.GetMessage()
	if c.rspToProto != nil {
		ret, err = c.rspToProto(rsp, err)
	}
	return ret, err
}

func (c *Client[T, H]) CallRawOnly(req T) error {
	bs, err := c.clientOption.GetReqCoder().Marshal(req)
	if err != nil {
		return err
	}
	c.RawSend(bs)
	return nil
}

func (c *Client[T, H]) RawSend(bs []byte) {
	c.writePacketBytesCh <- bs
}

func (c *Client[T, H]) run() {
	defer func() {
		c.cancel()
		c.conn.CloseConnectionAllCallback("exit run")
	}()
	c.wg.Add(4)
	go c.writeLoop()
	go c.readLoop()
	go c.tick()
	go c.process()
	select {
	case <-c.ctx.Done():
		return
	}
}

func (c *Client[T, H]) process() {
	defer c.wg.Done()
	dispatcher := c.clientOption.GetMessageDispatcher()
	defer c.cancel()
	for {
		select {
		case <-c.ctx.Done():
			return
		case m := <-c.receivedCh:
			c.cmr.Trigger(m)
			dispatcher.DispatchMessage(m, c.conn)
		}
	}
}

func (c *Client[T, H]) Stop() {
	c.cancel()
	c.wg.Wait()
}

func (c *Client[T, H]) writeLoop() {
	defer c.wg.Done()
	lowerWriteCh := c.conn.GetWriteCtx().(*inet.WriteToChanCtx).Buffer()
	rw := c.remoteOption.GetFrameReader()
	defer c.cancel()
	for {
		select {
		case <-c.ctx.Done():
			return
		case bs := <-lowerWriteCh:
			err := c.conn.LowerWrite(bs.Bytes())
			if err != nil {
				logx.Warnf("write to connection err %+v", err)
				rw.Return(bs)
				return
			}
			rw.Return(bs)
		case buf := <-c.writePacketBytesCh:
			err := c.remoteOption.GetFrameReader().WriteOnePacketBytes(c.conn, buf)
			if err != nil {
				logx.Errorf("write one packet err %+v", err)
			}
		}
	}
}

func (c *Client[T, H]) readLoop() {
	defer c.wg.Done()
	frameReaderWriter := c.remoteOption.GetFrameReader()
	rspCoder := c.clientOption.GetRspCoder()
	defer c.cancel()
	for {
		select {
		case <-c.ctx.Done():
			return
		default:
			packBytes, err := frameReaderWriter.BlockSlicePacketBytes(c.conn)
			if err != nil {
				logx.Errorf("read packBytes error %v", err)
				return
			}
			rsp, err := rspCoder.Unmarshal(packBytes)
			if err != nil {
				logx.Errorf("unmarshal packBytes error %v", err)
				continue
			}
			select {
			case c.receivedCh <- rsp.(H):
				continue
			default:
				logx.Warnf("tcpClient readOncePacket add to process queue failed")
				continue
			}
		}
	}
}

func (c *Client[T, H]) tick() {
	defer c.wg.Done()
	defer c.cancel()
	heartBeatTicker := time.NewTicker(c.clientOption.HeartBeatDuration())
	defer heartBeatTicker.Stop()
	for {
		select {
		case <-c.ctx.Done():
			return
		case <-heartBeatTicker.C:
			c.clientOption.GetHeartBeatFunc()(c)
		}
	}
}

func (c *Client[T, H]) GetCtx() context.Context {
	return c.ctx
}
